package com.zyx.flinkdemo.stream.utils;

import com.zyx.flinkdemo.pojo.Nested;
import com.zyx.flinkdemo.pojo.Props;
import com.zyx.flinkdemo.pojo.TransientSink;
import com.zyx.flinkdemo.stream.cons.CommonConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.util.List;

/**
 * @author zyx
 * @since 2021/6/8 23:17
 * desc: ClickHouse工具类
 */
public class ClickHouseUtil {
    public static <T> SinkFunction<T> getJdbcSink(String sql) {
        // obj  就是流中的一条数据对象
        return JdbcSink.sink(
                //要执行的SQL语句
                sql,
                // 执行写入操作   就是将当前流中的对象属性赋值给SQL的占位符 insert into visitor_stats_0820 values(?,?,?,?,?,?,?,?,?,?,?,?)
                (JdbcStatementBuilder<T>) (ps, obj) -> {
                    //获取当前类中  所有的属性
                    Field[] fields = obj.getClass().getDeclaredFields();
                    //跳过的属性计数
                    int skipOffset = 0;
                    for (int i = 0; i < fields.length; i++) {
                        Field field = fields[i];
                        //通过属性对象获取属性上是否有@TransientSink注解
                        TransientSink transientSink = field.getAnnotation(TransientSink.class);
                        //如果transientSink不为空，说明属性上有@TransientSink标记，那么在给?占位符赋值的时候，应该跳过当前属性
                        if (transientSink != null) {
                            skipOffset++;
                            continue;
                        }
                        //设置私有属性可访问
                        field.setAccessible(true);
                        try {
                            //获取属性值
                            Object o = field.get(obj);
                            ps.setObject(i + 1 - skipOffset, o);
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                //构建者设计模式，创建JdbcExecutionOptions对象，给batchSize属性赋值，执行执行批次大小
                new JdbcExecutionOptions.Builder().withBatchSize(5).build(),
                //构建者设计模式，JdbcConnectionOptions，给连接相关的属性进行赋值
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(CommonConfig.CLICKHOUSE_URL)
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build()
        );
    }


    public static <T> SinkFunction<T> getNestedJdbcSink(String sql) {
        // obj  就是流中的一条数据对象
        return JdbcSink.sink(
                //要执行的SQL语句
                sql,
                // 执行写入操作 就是将当前流中的对象属性赋值给SQL的占位符
                (JdbcStatementBuilder<T>) (ps, obj) -> {
                    // 获取当前类中  所有的属性
                    Field[] fields = obj.getClass().getDeclaredFields();
                    int j = 1;
                    for (Field field : fields) {
                        // 设置私有属性可访问
                        field.setAccessible(true);
                        if ("props".equals(field.getName())) {
                            Nested nested = (Nested) obj;
                            List<String[]> listStrArray = CommonUtils
                                    .listToStringArrayList(nested.getProps(), Props.class);
                            for (String[] strArray : listStrArray) {
                                ps.setArray(j++, ps.getConnection().createArrayOf("String", strArray));
                            }
                            continue;
                        }
                        try {
                            // 获取属性值
                            Object o = field.get(obj);
                            ps.setObject(j++, o);
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        }
                    }
                },
                //构建者设计模式，创建JdbcExecutionOptions对象，给batchSize属性赋值，执行执行批次大小
                new JdbcExecutionOptions.Builder().withBatchSize(5).build(),
                //构建者设计模式，JdbcConnectionOptions，给连接相关的属性进行赋值
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(CommonConfig.CLICKHOUSE_URL)
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .build()
        );
    }
}
